package com.flute.haflute.slave;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.rmi.RemoteException;
import java.util.List;
import java.util.Properties;

import org.jdom.JDOMException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.common.ClusterConstants;
import com.flute.haflute.common.ClusterContext;
import com.flute.haflute.common.ClusterConstants.CLI;
import com.flute.haflute.jobbox.base.RunProcessInterface;
import com.flute.haflute.jobbox.child.ChildContext;
import com.flute.haflute.jobbox.child.TaskReceiver;
import com.flute.haflute.jobbox.child.TaskTracker;
import com.flute.haflute.jobbox.common.CloudContext;
import com.flute.haflute.jobbox.common.RunProcessLauncher;
import com.flute.haflute.jobbox.communication.ChildCommunicationService;
import com.flute.haflute.monitor.GeneralStatusCenter;
import com.flute.haflute.monitor.JvmReport;
import com.flute.haflute.tools.ClusterUtils;

/**
 * 一项JOB的任务执行JVM。
 */
public class Child {
	private static Logger logger = LoggerFactory.getLogger(Child.class);
	private static final String logConfigFile = "conf/log-child.lcf";

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		try {
			System.out.println("Starting child, please wait...\n");
			initContext(args);
			startCommunicationServices(args);
			startChildBackends(args);
			startAgentService(args);
			starCustomBackends(args);
			addShutDownHook();
			setDefaultExceptionHander();
			new JvmReport(ChildContext.getMyID());
			// 启动时先向接口注册自己的名字
			GeneralStatusCenter.getInstance().setMyId(ChildContext.getMyID());

			System.out.println("\n====== Child started successfully, my ID : <"
					+ ChildContext.getMyID() + "> ======");
			System.out.println(ClusterConstants.PROCESS_STARTED);
			// redirectSystemOut(args);
		} catch (Exception e) {
			System.err.println("cloud Child will exit for ERROR: ");
			e.printStackTrace();
			ClusterUtils.exitGracefully();
		}
	}

	/**
	 * @param args
	 */
	private static void startAgentService(String[] args) {
		new ChildCommandHandler("Child Command Handler").start();
	}

	/**  */
	private static void setDefaultExceptionHander() {
		// 抓住没有捕获的异常，防止导致JVM崩溃
		Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
			public void uncaughtException(Thread t, Throwable e) {
				logger.error("Fatal exception in thread " + t, e);
				if (e instanceof OutOfMemoryError) {
					ClusterUtils.exitGracefully();
				}
			}
		});
	}

	/**  */
	private static void addShutDownHook() {
		// 注册子节点关闭时需要执行的Hook，以便清理环境
		Runtime.getRuntime().addShutdownHook(new ShutdownHook());
	}

	/** 初始化子节点系统、软件和平台环境 */
	private static void initContext(String[] args) throws JDOMException,
			IOException {
		Properties props = new Properties();
		for (String arg : args) {
			if (arg.startsWith(CLI.JOB_MASTER_IS)) {
				String master = arg.replaceFirst(CLI.JOB_MASTER_IS, "");
				props.setProperty(CLI.JOB_MASTER_IS, master);
				continue;
			}
			if (arg.startsWith(CLI.ROOTDIR_IS)) {
				String rootDir = arg.replaceFirst(CLI.ROOTDIR_IS, "");
				ClusterContext.setRootDir(rootDir);
				CloudContext.setRootDir(rootDir);
				logger.info("ROOT DIR is " + rootDir);
				continue;
			}
		}

		ClusterUtils.loadDefaultConfiguration();
		ClusterUtils.configLogger(logConfigFile);
		CloudContext.setCloudMaster(props.getProperty(CLI.JOB_MASTER_IS,
				CloudContext.getCloudMaster()));
		logger.info("settted master is " + CloudContext.getCloudMaster());
	}

	/** 启动子节点通信平台服务 */
	private static void startCommunicationServices(String[] args)
			throws RemoteException {
		System.out.println("begin to start Communication Service...");
		ChildCommunicationService service = new ChildCommunicationService();
		// 连接到主节点，并获取分配的全局唯一ID
		int myNo = service.connect2Master();
		ChildContext.setDaemonNo(String.valueOf(myNo));
		System.out.println("Communication Service started OK");
	}

	/**
	 * 启动用户自定义的后台线程
	 * 
	 * @throws FileNotFoundException
	 * @throws IllegalAccessException
	 * @throws InstantiationException
	 */
	private static void starCustomBackends(String... params) throws Exception,
			IllegalAccessException {
		// ClusterUtils.startCustomedBackends(params);
	}

	/** 所有的子节点关键后台线程在这里启动 */
	private static void startChildBackends(String... params) throws Exception {
		// 下面记录所有关键线程名字，便于后面打印完成消息
		String[] coreNames = new String[2];
		coreNames[0] = TaskTracker.class.getSimpleName();
		coreNames[1] = TaskReceiver.class.getSimpleName();

		// 真正启动后台线程
		RunProcessInterface[] coreBackends = new RunProcessInterface[2];
		coreBackends[0] = new TaskTracker(); // 后台执行调度线程
		coreBackends[1] = new TaskReceiver(); // 消息接收器
		for (RunProcessInterface process : coreBackends) {
			process.callMain(params);
			RunProcessLauncher.addRunProcess2Cache(process); // 放入全局Process池
		}

		// 检查所有的后台线程启动情况
		int initialized = 0;
		while (initialized < coreBackends.length) {
			for (int i = 0; i < coreBackends.length; i++) {
				if (coreBackends[i].isInitialized()) {
					initialized++;
					System.out.println("process [" + coreNames[i]
							+ "] started OK");
				}
			} // for( i..
			doWait();
		} // while( initialized..
	}

	/** 等待一定时间 */
	private static void doWait() {
		try {
			Thread.sleep(100L);
		} catch (Exception e) {
		}
	}

}

/**
 * 系统关闭前腰执行的钩子程序，清理环境
 * 
 * @author Liu Liu
 * @date 2010-6-2
 */
class ShutdownHook extends Thread {
	private static Logger logger = LoggerFactory.getLogger(ShutdownHook.class);

	@Override
	public void run() {
		logger.info("received exit signal, begin to shutdown...");
		// 通知所有Process关闭
		List<RunProcessInterface> processList = RunProcessLauncher
				.getRunProcessList();
		for (RunProcessInterface process : processList) {
			process.shutdown();
		}
		logger.info("shutdown all backend processes OK");
		logger.info("Job Child(Slave) shutdown OK");
	}
}